package org.atlantis.datasink.mx;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

final class FragmentMonitor {

	private final long monitorDurationNs;
	private final AtomicReference<Fragment> complete = new AtomicReference<Fragment>(new Fragment());
	private final AtomicReference<Fragment> current = new AtomicReference<Fragment>(new Fragment());

	FragmentMonitor() {
		this(TimeUnit.MINUTES.toNanos(1));
	}

	FragmentMonitor(long monitorDurationNs) {
		this.monitorDurationNs = monitorDurationNs;
	}

	void record() {
		Fragment frag = current.get();
		frag.add();
		long ns = System.nanoTime() - frag.start;
		if (ns >= monitorDurationNs) {
			boolean swapped = current.compareAndSet(frag, new Fragment());
			if (swapped) {
				complete.set(frag);
				frag.end.set(System.nanoTime());
			}
		}
	}

	double getSpeedBySec() {
		Fragment frag = complete.get();
		return frag.getTimes() / frag.durationSec();
	}

	double getSpeedByMs() {
		Fragment frag = complete.get();
		return frag.getTimes() / frag.durationMs();
	}

	private class Fragment {
		private static final double SEC2NANO = 1000.0 * 1000.0 * 1000.0;
		private static final double MS2NANO = 1000.0 * 1000.0;

		final long start = System.nanoTime();
		AtomicLong end = new AtomicLong(-1);

		private AtomicInteger times = new AtomicInteger();

		void add() {
			times.incrementAndGet();
		}

		int getTimes() {
			return times.get();
		}

		double durationSec() {
			return (end.get() - start) / SEC2NANO;
		}

		double durationMs() {
			return (end.get() - start) / MS2NANO;
		}
	}
}
